package com.leiyuee.flink.batch.kafka;

import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.connectors.kafka.KafkaDeserializationSchema;
import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.StringData;
import org.apache.flink.table.data.TimestampData;
import org.apache.flink.types.RowKind;
import org.apache.kafka.clients.consumer.ConsumerRecord;

import java.nio.charset.StandardCharsets;
import java.sql.Timestamp;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;

public class CustomerRowDataKafkaDeserialization implements KafkaDeserializationSchema<RowData> {

    @Override
    public boolean isEndOfStream(RowData nextElement) {
        return false;
    }

    @Override
    public RowData deserialize(ConsumerRecord<byte[], byte[]> record) throws Exception {
        // 创建 RowData 对象并填充数据
        GenericRowData row = null;
        if(record != null && record.value() != null) {
            String value = new String(record.value(), StandardCharsets.UTF_8);
            // record.topic() 的值为 cnhuam0itpoc85_EasyStar.dbo.sysDevice
            //String[] fields =  record.topic().split("\\.");
            //String database = fields[1];
            //String tableName = fields[2];

            // 将 value 的数据获取出来
            JSONObject valueJson = JSONObject.parseObject(value);

            // 根据数据 op 字段设置 RowKind，默认 INSERT
            RowKind rowKind = RowKind.INSERT;
            // 状态：增加[ insert ]、删除[ delete ]、修改[ update ]、查询[ read ]
            String type = valueJson.getOrDefault("op", null).toString().toLowerCase();
            // 需要写入的数据 Device 实体，默认获取 after 的数据
            Device target = JSONObject.toJavaObject(valueJson.containsKey("after") ? valueJson.getJSONObject("after") : null, Device.class);

            // RowKind，默认 INSERT，Device 实体默认获取 after 的数据，所以插入操作不需要判断
            if (type.equals("d") || type.equals("delete")) {
                // 数据删除，只包含 before 数据，RowKind 设置为 UPDATE_AFTER，Device 实体默认获取 before 的数据
                rowKind = RowKind.DELETE;
                target = JSONObject.toJavaObject(valueJson.containsKey("before") ? valueJson.getJSONObject("before") : null, Device.class);
            } else if (type.equals("u") || type.equals("update")) {
                // 数据修改，包含 before 和 after 数据，RowKind 设置为 UPDATE_AFTER，Device 实体默认获取 after 的数据
                rowKind = RowKind.UPDATE_AFTER;
            }


            row = new GenericRowData(16);
            row.setRowKind(rowKind);
            // 设置字段值
            row.setField(0, target.getID());  // ID
            String fType = target.getFType();
            StringData value1 = StringData.fromString(fType);
            row.setField(1, value1);  // FType
            row.setField(2, StringData.fromString(target.getDeviceNo()));  // DeviceNo
            row.setField(3, target.getUseType());  // UseType
            row.setField(4, StringData.fromString(target.getPlace()));  // Place
            row.setField(5, StringData.fromString(target.getIP()));  // IP
            row.setField(6, StringData.fromString(target.getNote()));  // Note
            row.setField(7, target.getUseType());  // GroupID
            row.setField(8, StringData.fromString(target.getCommPort()));  // CommPort
            row.setField(9, StringData.fromString(target.getDeviceType()));  // DeviceType
            row.setField(10, StringData.fromString(target.getUsername()));  // username
            row.setField(11, StringData.fromString(target.getPassword()));  // password
            row.setField(12, target.getStarttime() == null ? null : TimestampData.fromTimestamp(target.getStarttime()));  // starttime
            row.setField(13, target.getEndtime() == null ? null : TimestampData.fromTimestamp(target.getEndtime()));  // endtime

            // 获取当前时间
            LocalDateTime localDateTime = LocalDateTime.now();
            String partitionDay = localDateTime.format(DateTimeFormatter.ofPattern("yyyy-MM-dd"));
            // 这里添加缺失的字段值
            row.setField(14, TimestampData.fromTimestamp(new Timestamp(valueJson.getLong("ts_ms"))));  // ts
            row.setField(15, StringData.fromString(partitionDay));  // partition_day

            //System.out.printf("\n执行时间: %s\trecord 不为空，数据 [GenericRowData] 已返回: \n> %s \n",
            //        LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")), row);
        }// end of if(record != null && record.value() != null)
        return row;
    }

    @Override
    public TypeInformation<RowData> getProducedType() {
        return TypeInformation.of(RowData.class);
    }
}
